package kettle.plugin;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import one.util.VelocityUtils;

import org.apache.commons.io.FileUtils;
import org.apache.commons.vfs2.FileObject;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.core.vfs.KettleVFS;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

/**
 * 步骤类
 * @author	yaodh
 * @email	ydh110100@163.com	
 * @blog	http://www.cnblogs.com/lucky110100/
 */
public class VelocityStep extends BaseStep implements StepInterface {

	private VelocityStepData data;
	private VelocityStepMeta meta;
	private ArrayList<Map<String, Object>> rows = new ArrayList<Map<String, Object>>();
	
	public VelocityStep(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis) {
		super(s, stepDataInterface, c, t, dis);
	}

	@Override
	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
		meta = (VelocityStepMeta) smi;
		data = (VelocityStepData) sdi;

		Object[] r = getRow(); // get row, blocks when needed!
		if (r == null) // no more input to be expected...
		{
			setOutputDone();
			return false;
		}

		if (first) {
			first = false;

			data.outputRowMeta = getInputRowMeta().clone();
		    meta.getFields( data.outputRowMeta, getStepname(), null, null, this );

			logBasic("template step initialized successfully");

		}
		
		Map<String, Object> map = new HashMap<String, Object>();
		for (int c = 0; c < data.outputRowMeta.size(); c++) {
		      ValueMetaInterface v = data.outputRowMeta.getValueMeta(c);
		      String show;
		      show = v.getString(r[c]);
		      map.put(v.getName(), show);
		}
		rows.add(map);
		
		Object[] outputRow = RowDataUtil.addValueData(r, data.outputRowMeta.size() - 1, "dummy value");

		putRow(data.outputRowMeta, outputRow); // copy row to possible alternate rowset(s)

		if (checkFeedback(getLinesRead())) {
			logBasic("Linenr " + getLinesRead()); // Some basic logging
		}

		return true;
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (VelocityStepMeta) smi;
		data = (VelocityStepData) sdi;

		return super.init(smi, sdi);
	}

	public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (VelocityStepMeta) smi;
		data = (VelocityStepData) sdi;
		
		try {
			Map<String, Object> context = new HashMap<String, Object>();
			context.put("rows", rows);
			if(meta.isUseVariableSubstitution()){
				//将kettle所有变量加入模板变量中
				String[] varArr = listVariables();
				for(String var : varArr){
					context.put(var, getVariable(var));
				}
			}
			
			//封装模板的模板
			if(!meta.isVmofvmfromfile()){
				context.put("childvm", meta.getVmofvm());
			}else{
				String vmofvmfilename = environmentSubstitute(meta.getVmofvmfile());
				FileObject vmofvmfile = KettleVFS.getFileObject(vmofvmfilename, getTransMeta());
				String svmofvm = FileUtils.readFileToString(new File(vmofvmfile.getURL().getFile()));
				context.put("childvm", svmofvm);
			}
			
			//替换环境变量
			String outfilename = environmentSubstitute(meta.getOutfilename());
			FileObject outfile = KettleVFS.getFileObject(outfilename, getTransMeta());
			
			if(!meta.isSqlfromfile()){
				VelocityUtils.mergeStrToFile(meta.getSql(), context, outfile.getURL().getFile(), meta.isAppend());
			}else{
				String vmfilename = environmentSubstitute(meta.getSqlfilename());
				FileObject vmfile = KettleVFS.getFileObject(vmfilename, getTransMeta());
				VelocityUtils.mergeToFile(vmfile.getURL().getFile(), context, outfile.getURL().getFile(), meta.isAppend());
			}
			
		} catch (IOException e) {
			logError(e.toString());
            setErrors(1);
		} catch (Exception e) {
			logError(e.toString());
			setErrors(1);
		}
		super.dispose(smi, sdi);
	}


}
